/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.columndef;

import edu.ucsd.db.bassarisk.exception.SchemaValidationException;
import edu.ucsd.db.bassarisk.query.path.DSDBuilder;
import edu.ucsd.db.bassarisk.query.path.DSDEdgeBuilder;
import edu.ucsd.db.bassarisk.query.path.DSDElements.DSDAggregate;
import edu.ucsd.db.bassarisk.query.path.DSDElements.DSDDescent;
import edu.ucsd.db.bassarisk.query.path.DSDElements.DSDElement;
import edu.ucsd.db.bassarisk.query.path.DataSequenceDefinition;
import edu.ucsd.db.bassarisk.schema.Schema;
import edu.ucsd.db.bassarisk.schema.Schema.ValueType;
import edu.ucsd.db.bassarisk.schema.SchemaNode;
import edu.ucsd.db.bassarisk.treepattern.EdgeFunction;

public class DataColumnDefinition extends DataSequenceDefinition {
  private final int size;
  private final DSDDescent aggregationContext;
  private final DSDDescent aggregationTarget;
  private final DSDDescent aggregation;
  private EdgeFunction aggregationFunction;
  private SchemaNode targetSequence;

  public static class DataColumnDefinitionBuilder extends
      DSDBuilder<DataColumnDefinitionBuilder> {
    private int size;
    private DSDDescent last;
    private DSDDescent aggregation;
    private DSDDescent aggregationContext;

    private DataColumnDefinitionBuilder(final ValueType type, final int size) {
      super(type);
      this.size = size;
    }

    @Override
    protected void appendElement(DSDElement elem) {
      if (elem instanceof DSDAggregate
          || (elem instanceof DSDDescent && ((DSDDescent) elem).isFilter())) {
        if (this.aggregation != null) {
          String msg = "Multiple aggregations in column definition.";
          throw new RuntimeException(msg);
        }
        this.aggregation = (DSDDescent) elem;
        this.aggregationContext = this.last;
      }
      if (elem instanceof DSDDescent) {
        this.last = (DSDDescent) elem;
      }
      super.appendElement(elem);
    }

    public DataColumnDefinition toDefinition() {
      return new DataColumnDefinition(this.getHead(), this.type, this.size,
                                      this.aggregationContext, this.aggregation,
                                      this.last);
    }
  }

  public static DSDEdgeBuilder<DataColumnDefinitionBuilder>
      fromRoot(final ValueType type, int size) {
    return new DataColumnDefinitionBuilder(type, size);
  }

  private DataColumnDefinition(final DSDElement head, final ValueType type,
                               final int size, DSDDescent aggContext,
                               DSDDescent aggregation, DSDDescent aggTarget) {
    super(head, type);
    this.size = size;
    this.aggregationContext = aggContext;
    this.aggregation = aggregation;
    this.aggregationTarget = aggTarget;
  }

  public void initialize(Schema schema) throws SchemaValidationException {
    if (this.aggregation != null) {
      if (this.aggregation instanceof DSDAggregate) {
        this.aggregationFunction = ((DSDAggregate) this.aggregation).getFunction();
      } else {
        this.aggregationFunction = EdgeFunction.NonEmpty;
      }
    } else {
      this.aggregationFunction = EdgeFunction.Values;
    }
    int size = this.aggregationTarget.getSequenceClass().getRefinement().size();
    if (size == 1) {
      this.targetSequence =
          (SchemaNode) this.aggregationTarget.getSequenceClass().getRefinement()
              .toArray()[0];
    } else {
      String empty =
          "A column in the database is found to not match the supplied "
              + "schema definition.  Please check your schema. ";
      String duplicates =
          "Ambiguous database column found.  Database columns should "
              + "contain data from exactly one schema sequence.";
      String msg = size == 0 ? empty : duplicates;
      throw new SchemaValidationException(msg);
    }
  }

  public DSDDescent getAggregationContext() {
    return this.aggregationContext;
  }

  public DSDDescent getAggregationTarget() {
    return this.aggregationTarget;
  }

  public EdgeFunction aggregationFunction() {
    return this.aggregationFunction;
  }

  public int cost() {
    return this.size;
  }

  public SchemaNode getTargetSequence() {
    return this.targetSequence;
  }

  public Object lengthEntries() {
    return this.size;
  }
}
